/***
 * Copyright (c) 2021-2031 murenchao
 * fig is licensed under Mulan PubL v2.
 * You can use this software according to the terms and conditions of the Mulan PubL v2.
 * You may obtain a copy of Mulan PubL v2 at:
 *       http://license.coscl.org.cn/MulanPubL-2.0
 * THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND,
 * EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT,
 * MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
 * See the Mulan PubL v2 for more details.
 */
package cool.taomu.software.fig.mqtt.broker.impl.request

import cool.taomu.software.fig.classloader.FigClassLoaderManage.Autowired
import cool.taomu.software.fig.configure.ConfigureManage
import cool.taomu.software.fig.mqtt.broker.MQTTCache
import cool.taomu.software.fig.mqtt.broker.entity.MessageEntity
import cool.taomu.software.fig.mqtt.broker.entity.PubAckEntity
import cool.taomu.software.fig.mqtt.broker.impl.IRpcPublish
import cool.taomu.software.fig.mqtt.broker.impl.PublishObservable
import cool.taomu.software.fig.mqtt.broker.inter.IPublishObserver.Type
import cool.taomu.software.fig.mqtt.broker.inter.IRequest
import cool.taomu.software.fig.mqtt.broker.inter.IResponse
import cool.taomu.software.fig.mqtt.utils.CommonUtils
import cool.taomu.software.fig.rpc.RpcProxy
import io.netty.channel.ChannelHandlerContext
import io.netty.handler.codec.mqtt.MqttMessage
import io.netty.handler.codec.mqtt.MqttPublishMessage
import io.netty.util.ReferenceCountUtil
import org.slf4j.LoggerFactory
import io.netty.handler.codec.mqtt.MqttQoS

/**
 *  PUBLISH 	|	3 	|	两个方向都允许 	|	发布消息
 */
class MQTTPublish implements IRequest {
    val LOG = LoggerFactory.getLogger(MQTTPublish)

    @Autowired(cool.taomu.software.fig.mqtt.broker.impl.response.MQTTPubAck)
    IResponse<PubAckEntity> ack;

    MQTTCache cache = MQTTCache.instance;
    val config = ConfigureManage.loadConfig;

    override request(ChannelHandlerContext ctx, MqttMessage mqttMessage) {
        try {
            var publishMessage = mqttMessage as MqttPublishMessage;
            var message = new MessageEntity()
            message.senderId = CommonUtils.getClientId(ctx.channel())
            LOG.info("执行了MQTT Publish 命令 : " + message.senderId);
            message.qos = publishMessage.fixedHeader().qosLevel();
            message.topic = publishMessage.variableHeader().topicName();
            message.payload = (mqttMessage as MqttPublishMessage).payload();
            message.type = mqttMessage.fixedHeader().messageType().value();
            message.dup = publishMessage.fixedHeader().isDup();
            message.retain = publishMessage.fixedHeader().isRetain();
            message.msgId = publishMessage.variableHeader().packetId();
            message.senderChannel = ctx.channel;
            switch (message.qos) {
                case EXACTLY_ONCE,
                case AT_MOST_ONCE: {
                    LOG.info(String.format("Qos0 message,clientId=%s", message.senderId));
                    retainMessage(message);
                }
                case AT_LEAST_ONCE: {
                    LOG.info(String.format("Qos1 message,clientId=%s", message.senderId));
                    retainMessage(message);
                    var messageId = message.msgId;
                    var entity = new PubAckEntity(messageId);
                    // 回给发送者
                    return #[ack.response(entity)];
                }
                default: {
                    LOG.info(String.format("Wrong mqtt message,clientId=%s", message.senderId));
                }
            }
        } catch (Exception ex) {
            ex.printStackTrace
        } finally {
            ReferenceCountUtil.release(mqttMessage.payload());
        }
        return null;
    }

    protected synchronized def retainMessage(MessageEntity message) {
        LOG.debug(
            "clientId 为 {} 是否存在 Retain 数据 {}, 接受到的数据为 {} ",
            message.senderId,
            message.retain,
            new String(message.payload)
        );
        cache.store(message.senderId, message);
        // 需要功能验证
        if (message.retain) {
            var qos = message.qos;
            var payload = message.payload;
            // msgVal.clientId 为发送者的clientId
            if (qos === MqttQoS.AT_MOST_ONCE || payload === null || payload.length === 0) {
                LOG.info("清空 clientId 为 {} 的Retain数据", message.senderId);
                cache.clearRetainBySenderId(message.senderId);
            } else {
                LOG.info("保存 clientId 为 {} 的Retain数据", message.senderId);
                cache.storeRetain(message.senderId, message);
                CommonUtils.exec([
                    config.mqtt.cluster.filterNull.forEach [ addr |
                        CommonUtils.exec([
                            var rpc = RpcProxy.<IRpcPublish>getInstance(addr.ip, addr.port, IRpcPublish);
                            rpc.refreshRetain();
                        ]);
                    ]
                ]);
            }
        }
        // 推送用户订阅
        PublishObservable.instance.start(message.senderId, Type.MESSAGE);
        CommonUtils.exec([
            config.mqtt.cluster.filterNull.forEach [ addr |
                CommonUtils.exec([
                    var rpc = RpcProxy.<IRpcPublish>getInstance(addr.ip, addr.port, IRpcPublish);
                    rpc.publish(message);
                ]);
            ]
        ]);
    }
}
